package com.wwjd.starter.canal.client.core;

import com.alibaba.otter.canal.client.CanalConnector;
import com.wwjd.starter.canal.annotation.ListenPoint;
import com.wwjd.starter.canal.client.abstracts.AbstractCanalClient;
import com.wwjd.starter.canal.client.interfaces.CanalEventListener;
import com.wwjd.starter.canal.client.interfaces.MessageTransponder;
import com.wwjd.starter.canal.config.CanalProperties;
import com.wwjd.starter.canal.util.ApplicationContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.springframework.core.annotation.AnnotatedElementUtils;

import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.*;

/**
 * 通过线程池处理
 *
 * @author 阿导
 * @CopyRight 萬物皆導
 * @created 2018/5/28 14:52
 * @Modified_By 阿导 2018/5/28 14:52
 */
@Slf4j
public class SimpleCanalClient extends AbstractCanalClient {
    /**
     * 声明一个线程池
     */
    private ThreadPoolExecutor executor;

    /**
     * 通过实现接口的监听器
     */
    protected final List<CanalEventListener> listeners = new ArrayList<>();

    /**
     * 通过注解的方式实现的监听器
     */
    private final List<ListenerPoint> annoListeners = new ArrayList<>();

    /**
     * 构造方法，进行一些基本信息初始化
     *
     * @param canalProperties
     * @return
     * @author 阿导
     * @time 2018/5/28 15:33
     * @CopyRight 万物皆导
     */
    public SimpleCanalClient(CanalProperties canalProperties) {
        super(canalProperties);
        executor = new ThreadPoolExecutor(5, 20, 120L, TimeUnit.SECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory());
        // 获取接口监听器
        List<CanalEventListener> list = ApplicationContextUtils.getBeansOfType(CanalEventListener.class);
        if (CollectionUtils.isNotEmpty(list)) {
            listeners.addAll(list);
        }
        // 获取注解监听器
        Map<String, Object> listenerMap = ApplicationContextUtils.getBeansWithAnnotation(com.wwjd.starter.canal.annotation.CanalEventListener.class);
        if (MapUtils.isNotEmpty(listenerMap)) {
            for (Object target : listenerMap.values()) {
                // 方法获取
                Method[] methods = target.getClass().getDeclaredMethods();
                if (methods != null && methods.length > 0) {
                    for (Method method : methods) {
                        // 获取监听的节点：AnnotatedElementUtils 支持子注解覆盖父注解的属性，而 AnnotationUtils 则不可以
                        ListenPoint l = AnnotatedElementUtils.findMergedAnnotation(method, ListenPoint.class);
                        if (l != null) {
                            annoListeners.add(new ListenerPoint(target, method, l));
                        }
                    }
                }
            }
        }
        if (log.isDebugEnabled() && CollectionUtils.isEmpty(listeners) && CollectionUtils.isEmpty(annoListeners)) {
            log.warn("该项目中没有任何监听的目标! ");
        }
        log.info("监听器初始化完成...");
    }

    /**
     * @param connector
     * @param config
     * @return
     * @author 阿导
     * @time 2018/5/28 15:01
     * @CopyRight 万物皆导
     */
    @Override
    protected void process(CanalConnector connector, Map.Entry<String, CanalProperties.Instance> config) {
        MessageTransponder messageTransponder = factory.newTransponder(connector, config, listeners, annoListeners);
        executor.submit(messageTransponder);
    }

    /**
     * 关闭 canal 客户端
     *
     * @param
     * @return
     * @author 阿导
     * @time 2018/5/28 15:00
     * @CopyRight 万物皆导
     */
    @Override
    public void stop() {
        // 停止 canal 客户端
        super.stop();
        // 线程池关闭
        executor.shutdown();
    }
}
